package com.apes.framework.plugin.mq.rocketmq.autoconfigure;

import com.apes.framework.plugin.esb.api.SimpleRequest;
import com.apes.framework.util.EventUtil;
import com.apes.framework.util.MapUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.rocketmq.common.admin.ConsumeStats;
import org.apache.rocketmq.common.admin.OffsetWrapper;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.body.ConsumeStatsList;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

@Component
public class RocketMQAdmin {
    protected final Logger logger = LogManager.getLogger(RocketMQAdmin.class);
    private DefaultMQAdminExt defaultMQAdminExt;

    @Value("${rocketmq.name-server:}")
    private String nameServer;

    private String brokerAddr;

    public RocketMQAdmin() {
        /**初始化监控客户端**/
        try {
            defaultMQAdminExt = new DefaultMQAdminExt();
            defaultMQAdminExt.setInstanceName(Long.toString(System.currentTimeMillis()));
            defaultMQAdminExt.start();
        } catch (Exception e) {
            logger.error("监控客户端启动失败...", e);
        }
        /**当jvm关闭的时,关闭客户端**/
        Runtime.getRuntime().addShutdownHook(new Thread(() -> defaultMQAdminExt.shutdown()));
    }

    private String getBrokerAddr() {
        if (StringUtils.isEmpty(nameServer)) return null;
        String[] urls = nameServer.split(":");
        if (urls.length == 0) return null;
        return urls[0] + ":" + "10909";
    }

    public void health(SimpleRequest request) {
        try {
            int size = request.get("size");
            String users = request.get("users");
            boolean isOrder = false;
            long timeoutMillis = 50000;
            long diffLevel = 10;
            ConsumeStatsList consumeStatsList = defaultMQAdminExt.fetchConsumeStatsInBroker(getBrokerAddr(), isOrder, timeoutMillis);
            if (consumeStatsList.getTotalDiff() < size) return;
            Map<String, Map<String, Long>> results = new HashMap<>();
            for (Map<String, List<ConsumeStats>> map : consumeStatsList.getConsumeStatsList()) {
                for (Map.Entry<String, List<ConsumeStats>> entry : map.entrySet()) {
                    String group = entry.getKey();
                    long subtotal = 0;
                    List<ConsumeStats> consumeStatsArray = entry.getValue();
                    for (ConsumeStats consumeStats : consumeStatsArray) {
                        List<MessageQueue> mqList = new LinkedList<>();
                        mqList.addAll(consumeStats.getOffsetTable().keySet());
                        for (MessageQueue mq : mqList) {
                            OffsetWrapper offsetWrapper = consumeStats.getOffsetTable().get(mq);
                            long diff = offsetWrapper.getBrokerOffset() - offsetWrapper.getConsumerOffset();
                            if (diff < diffLevel) {
                                continue;
                            }
                            subtotal += diff;
                        }
                    }
                    setValue(results, group, subtotal);
                }
            }
            String content = format(results);
            EventUtil.sendSynEvent(this, "event:dingding.message.publish", MapUtil.mapper(
                    "userList", users,
                    "content", "RocketMQ 有：" + consumeStatsList.getTotalDiff() + " 条消息待消费！" + "\n" + content
            ));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void setValue(Map<String, Map<String, Long>> results, String group, long subtotal) {
        String[] keys = group.split("%");
        if (subtotal > 0 && keys.length == 2) {
            String key = keys[0];
            Map values;
            if (results.containsKey(key)) {
                values = results.get(key);
            } else {
                values = new HashMap();
                results.put(key, values);
            }
            values.put(keys[1], subtotal);
        }
    }

    private String format(Map<String, Map<String, Long>> results) {
        StringBuffer sb = new StringBuffer();
        results.forEach((key, values) -> {
            if (sb.length() > 0) {
                sb.append("\n");
            }
            sb.append(key).append(" ").append("{");
            values.forEach((k, v) -> {
                sb.append("\n").append("  ").append(k.replace("_", ".")).append(":").append(" ").append(v);
            });
            sb.append("\n").append("}");
        });
        return sb.toString();
    }
}
